﻿#include <stdio.h>
#include <stdlib.h>
#include<sys/time.h>
#include "async_task.h"

#define LOG_ERR printf
#define LOG_DBG printf


typedef struct _async_thread
{
    pthread_t     thread_id;    // 线程id
    volatile int  req_abort;    // 线程退出标志，=1时请求退出
    int64_t timer_precision;    // 无执行任务时休眠间隔
    pthread_mutex_t mutex;
    int     count;              // 任务计数器，分配任务ID
    priority_queue_t *all_tasks_queue;      // 任务队列
    priority_queue_t *todo_tasks_queue;     // 要执行的任务队列,从all_tasks_queue中挑选
    priority_queue_t *req_del_tasks_queue;  // 请求删除队列
}async_thread_t;


typedef struct  _task_info
{
    task_type_e task_type;      // 任务类型
    uint64_t execute_time;      // 记录下一次执行的时间
    uint64_t interval_time;     // 间隔时间
    void *user_data;            // 用户指针
    task_ptr_t task_func_ptr;   // 任务函数指针
    int id;                     // 任务ID，由线程分配
}task_info_t;


/**
 * @brief 获取当前时间，单位为毫秒
 * @return 返回当前时间，单位为毫秒
 */
uint64_t get_now_time()
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec * 1000 + tv.tv_usec / 1000;
}

/**
 * @brief 将需要执行的任务放入todo_tasks_queue执行队列
 * @param src   任务指针
 * @param user_data
 */
static void select_todo_tasks_callback(void *src, void *user_data)
{
    int64_t now_time;
    async_thread_t *async_thread = (async_thread_t *)user_data;
    task_info_t *task_info = (task_info_t *)src;
    now_time = get_now_time();

    //    LOG_DBG("select_todo_tasks_callback task_id = %d, time = %ld\n",task_info->id, (int64_t)(now_time/1000));

    if((task_info->execute_time < now_time) ||          // 定时任务是否到达执行时间
            (task_info->task_type == kOnceOffTask))     // 一次性任务
    {
        task_info->execute_time = now_time + task_info->interval_time;      // 下一次的执行时间
        priority_queue_push(async_thread->todo_tasks_queue, task_info);        // 插入队列
    }
}

/**
 * @brief 线程循环，该函数可重入
 * @param user_data 线程句柄
 * @return
 */
static void *thread_loop(void* user_data)
{
    async_thread_t *async_thread = (async_thread_t *)user_data;
    task_info_t *task_info;

//    LOG_DBG("%s(%d) time = %ld\n", __FUNCTION__, __LINE__, (int64_t)(get_now_time()/1000));
    time_t timep;
    time(&timep); /*获取time_t类型当前时间*/
    /*转换为常见的字符串：Fri Jan 11 17:04:08 2008*/
    printf("thread_loop into, time = %s\n", ctime(&timep));

    while (1)         // 线程
    {
        // 1. 检查是否需要退出线程
        if(async_thread->req_abort == 1)
        {
            LOG_DBG("%s(%d) request abort thread\n", __FUNCTION__, __LINE__);
            break;
        }

        // 2. 检查任务队列是否有任务
        if(priority_queue_is_empty(async_thread->all_tasks_queue))
        {
            usleep(async_thread->timer_precision * 1000);
            continue;
        }
        // 3. 检查是否需要删除任务，这里会将req_del_tasks_queue的所有元素都取出来
        while(priority_queue_is_empty(async_thread->req_del_tasks_queue) != 1)
        {
            task_info = (task_info_t *)priority_queue_pop_head(async_thread->req_del_tasks_queue);
            LOG_DBG("%s(%d) task id = %d request delete it\n", __FUNCTION__, __LINE__,
                    task_info->id);
            if(priority_queue_find(async_thread->all_tasks_queue, task_info))
            {
                // 从任务队列中删除
                if(priority_queue_delete_item(async_thread->all_tasks_queue, (void *)task_info) < 0)
                    LOG_ERR("%s(%d) delete item failed, please check it\n", __FUNCTION__, __LINE__);

                // 释放task_info资源
                free(task_info);        // 只有在all_tasks_queue查找到才做释放task_info的操作
                task_info = NULL;
            }
            else
            {
                LOG_DBG("%s(%d) task id = %d no in the task queue, it have deleted\n", __FUNCTION__, __LINE__,
                        task_info->id);
            }
        }

        // 4. 从任务队列获取需要执行的任务
        //        LOG_DBG("task queue size = %d\n", priority_queue_size(async_thread->all_tasks_queue));
        priority_queue_traverse(async_thread->all_tasks_queue, async_thread, select_todo_tasks_callback);
        if(priority_queue_is_empty(async_thread->todo_tasks_queue))
        {
            usleep(async_thread->timer_precision * 1000);       // 如果没有可以执行的任务则休眠
            continue;
        }

        while (priority_queue_is_empty(async_thread->todo_tasks_queue) != 1)
        {
            task_info = (task_info_t *)priority_queue_pop_head(async_thread->todo_tasks_queue);
            //            LOG_DBG("%s(%d) task_info = %p, %p, %s\n", __FUNCTION__, __LINE__,
            //                    task_info, task_info->task_func_ptr, (char *)task_info->user_data);
            // 先查找是否在删除队列
            if(priority_queue_find(async_thread->req_del_tasks_queue, task_info))
            {
                // 如果请求删除队列中也有该任务句柄，则不执行回调函数并且将其从task queue中删除
                LOG_DBG("%s(%d) task id = %d need delete it immediately\n",
                        __FUNCTION__, __LINE__, task_info->id);
                // 从任务队列中删除
                if(priority_queue_delete_item(async_thread->all_tasks_queue, (void *)task_info) < 0)
                    LOG_ERR("%s(%d) delete item failed, please check it\n", __FUNCTION__, __LINE__);
                // 从请求删除队列中删除
                if(priority_queue_delete_item(async_thread->req_del_tasks_queue, (void *)task_info) < 0)
                    LOG_ERR("%s(%d) delete item failed, please check it\n", __FUNCTION__, __LINE__);
                // 释放task_info资源
                free(task_info);            // 只有在all_tasks_queue查找到才做释放task_info的操作
                task_info = NULL;
            }
            else
            {
                // 不在删除队列，所以执行回调函数
                task_info->task_func_ptr(task_info->user_data);
                // 执行完回调函数后则判断该任务是否是循环任务
                if(task_info->task_type == kOnceTimerTask || task_info->task_type == kOnceOffTask)
                {
                    LOG_DBG("%s(%d) task id = %d is once time task, so delete it from task queue\n",
                            __FUNCTION__, __LINE__, task_info->id);
                    // 一次性任务则从任务队列中删除，这里删除返回值为-1也不代表出错
                    if(priority_queue_delete_item(async_thread->all_tasks_queue, (void *)task_info) < 0)
                        LOG_ERR("%s(%d) delete item failed, please check it\n", __FUNCTION__, __LINE__);
                    // 释放task_info资源
                    free(task_info);
                    task_info = NULL;
                }
            }
        }
    }
    LOG_DBG("%s(%d) thread id = %lu, thread_loop finish\n",
            __FUNCTION__, __LINE__, (uint64_t)async_thread->thread_id);
    return ((void*)0);
}

/**
 * @brief 初始化一个异步线程
 * @param timer_precision 线程调度任务的时间精度，单位为毫秒，当没有可以执行的任务时先休眠timer_precision毫秒再去检测任务队列
 * @return 返回操作异步线程的句柄
 */
thread_handle_t aysnc_task_init(int64_t timer_precision)
{
    int ret;
    // 分配async_thread_t
    async_thread_t *async_thread = (async_thread_t *)malloc(sizeof(async_thread_t));
    if(!async_thread)
    {
        LOG_ERR("%s(%d) malloc async_thread_t failed\n", __FUNCTION__, __LINE__);
        return NULL;
    }
    // 初始化队列
    async_thread->all_tasks_queue = priority_queue_init();
    if(!async_thread->all_tasks_queue)
    {
        LOG_ERR("%s(%d) init all_tasks_queue failed\n", __FUNCTION__, __LINE__);
        goto all_tasks_queue_failed;
    }
    async_thread->todo_tasks_queue = priority_queue_init();
    if(!async_thread->todo_tasks_queue)
    {
        LOG_ERR("%s(%d) init todo_tasks_queue failed\n", __FUNCTION__, __LINE__);
        goto todo_tasks_queue_failed;
    }
    async_thread->req_del_tasks_queue = priority_queue_init();
    if(!async_thread->req_del_tasks_queue)
    {
        LOG_ERR("%s(%d) init req_del_tasks_queue failed\n", __FUNCTION__, __LINE__);
        goto req_del_tasks_queue_failed;
    }
    async_thread->req_abort = 0;
    // 定时器精度
    async_thread->timer_precision = timer_precision;

    // 初始化mute
    if(pthread_mutex_init(&async_thread->mutex, NULL) < 0)
    {
        LOG_ERR("%s(%d) pthread_mutex_init failed\n", __FUNCTION__, __LINE__);

        goto mutex_init_failed;
    }

    // 创建线程
    if ((ret = pthread_create(&async_thread->thread_id, 0, thread_loop, async_thread)) < 0)
    {
        LOG_ERR("Failed to spawn thread = %d \n", ret);
        goto pthread_create_failed;
    }
    // 启动线程
    return (thread_handle_t)async_thread;

pthread_create_failed:
    pthread_mutex_destroy(&async_thread->mutex);
mutex_init_failed:
    priority_queue_deinit(async_thread->req_del_tasks_queue);
req_del_tasks_queue_failed:
    priority_queue_deinit(async_thread->todo_tasks_queue);
todo_tasks_queue_failed:
    priority_queue_deinit(async_thread->all_tasks_queue);
all_tasks_queue_failed:
    free(async_thread);
    async_thread = NULL;

    return (thread_handle_t)async_thread;
}

/**
 * @brief 释放异步线程，如果还有任务则销毁任务队列
 * @param thread_handle 操作异步线程的句柄
 */
void aysnc_task_deinit(thread_handle_t thread_handle)
{
    async_thread_t *async_thread = (async_thread_t *)thread_handle;

    task_info_t *task_info;
    if(!async_thread)
    {
        LOG_ERR("async_thread is NULL\n");
        return;
    }
    void* stat;
    async_thread->req_abort = 1;
    LOG_DBG("%s(%d) pthread_join into\n", __FUNCTION__, __LINE__);
    int ret = pthread_join(async_thread->thread_id, &stat);
    if (ret != 0)
    {
        LOG_ERR("Internal error: pthread_join() returned %d\n", ret);
    }
    // 清空队列
    LOG_DBG("%s(%d) clear req_del_tasks_queue\n", __FUNCTION__, __LINE__);
    while(priority_queue_is_empty(async_thread->req_del_tasks_queue) != 1)
    {
        task_info = (task_info_t *)priority_queue_pop_head(async_thread->req_del_tasks_queue);
        LOG_DBG("%s(%d)  req_del_tasks_queue task id = %d\n", __FUNCTION__, __LINE__, task_info->id);
    }
    priority_queue_deinit(async_thread->req_del_tasks_queue);

    LOG_DBG("%s(%d) clear todo_tasks_queue\n", __FUNCTION__, __LINE__);
    while(priority_queue_is_empty(async_thread->todo_tasks_queue) != 1)
    {
        task_info = (task_info_t *)priority_queue_pop_head(async_thread->todo_tasks_queue);
        LOG_DBG("%s(%d)  todo_tasks_queue task id = %d\n", __FUNCTION__, __LINE__, task_info->id);

    }
    priority_queue_deinit(async_thread->todo_tasks_queue);

    LOG_DBG("%s(%d) clear all_tasks_queue\n", __FUNCTION__, __LINE__);
    while(priority_queue_is_empty(async_thread->all_tasks_queue) != 1)
    {
        task_info = (task_info_t *)priority_queue_pop_head(async_thread->all_tasks_queue);
        LOG_DBG("%s(%d)  all_tasks_queue task id = %d\n", __FUNCTION__, __LINE__, task_info->id);
        free(task_info);
    }
    priority_queue_deinit(async_thread->all_tasks_queue);

    if(pthread_mutex_destroy(&async_thread->mutex) < 0)
    {
        LOG_ERR( "%s(%d) pthread_mutex_destroy error\n" , __FUNCTION__, __LINE__);
    }
    free(async_thread);
    LOG_DBG("aysnc_task_deinit finish\n");
}

/**
 * @brief 异步执行任务
 * @param async_thread 线程句柄
 * @param task_type 任务类型
 * @param interval_time 间隔执行时间，单位为毫秒
 * @param user_data 任务参数
 * @param task_func_ptr  任务函数指针
 * @return 返回任务句柄，调用者可以根据该句柄去删除任务
 */
task_handle_t async_task_add(thread_handle_t thread_handle,
                             task_type_e task_type,
                             uint64_t interval_time,
                             void *user_data,
                             task_ptr_t task_func_ptr
                             )
{
    async_thread_t *async_thread = (async_thread_t *)thread_handle;
    // 判断参数的有效性
    if(!async_thread || !task_func_ptr)
    {
        LOG_ERR("async_thread or task_func_ptr is NULL\n");
        return NULL;
    }

    // 创建task对象
    task_info_t *task_info = (task_info_t *)malloc(sizeof(task_info_t));
    if(!task_info)
    {
        LOG_ERR("malloc task_info_t failed\n");
        return NULL;
    }
    task_info->task_type = task_type;       // 任务类型
    task_info->execute_time = interval_time + get_now_time();       // 根据当前时间以及需要间隔的时间计算下一次执行时间
    task_info->interval_time = interval_time;
    task_info->user_data = user_data;       // 用户指针
    task_info->task_func_ptr = task_func_ptr;   // 任务执行函数
    pthread_mutex_lock( &async_thread->mutex );
    task_info->id = async_thread->count++;
    pthread_mutex_unlock( &async_thread->mutex );

    LOG_DBG("\n ------ %s(%d) id = %d ------------\n\n", __FUNCTION__, __LINE__, task_info->id);
    priority_queue_push(async_thread->all_tasks_queue, task_info);

    return (task_handle_t)task_info;
}

void async_task_delete(thread_handle_t thread_handle, task_handle_t task_handle)
{
    async_thread_t *async_thread = (async_thread_t *)thread_handle;

    if(!async_thread || !task_handle)
    {
        LOG_ERR("async_thread or task_handle is NULL\n");
        return;
    }
    priority_queue_push(async_thread->req_del_tasks_queue, task_handle);  // 加入取消队列
}
